package top.infopub.concurrent;


import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import top.infopub.zookeeper.EvnUtils;


/**
 * using increasement threadfactory
 * @author Awoke
 * @version 2018年9月13日
 * @see ExecutorManager
 * @since
 */
public class ExecutorManager {

    private static ExecutorService exec = null;

    private static Object lock = new Object();

    private static Logger LOGGER = LoggerFactory.getLogger(ExecutorManager.class);

    private ExecutorManager() {

    }

    /** 线程池大小/每个cpu **/
    private static int POOL_THREADS_PER_PROCESSOR = 5;

    private static ExecutorManager instance = new ExecutorManager();

    static {
        String size = EvnUtils.getValue("pool.threads.per.processor");
        if (StringUtils.isNotBlank(size)) {
            POOL_THREADS_PER_PROCESSOR = Integer.valueOf(size);
        }
    }

    /**
     * Get Instance
     * @return 
     * @see
     */
    public static ExecutorManager getInstance() {
        synchronized (lock) {
            if (exec == null) {
                // IncreaseNameThreadFactory
                ThreadFactory threadFactory = new IncreaseNameThreadFactory(ExecutorManager.class);
                exec = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()
                                                    * POOL_THREADS_PER_PROCESSOR, threadFactory);
            }
        }
        return instance;
    }

    /**
     * 关闭exec
     * @throws Exception 
     * @see
     */
    public static void shutdown()
        throws Exception {
        if (exec != null) {
            LOGGER.debug("ExecutorService Closing...");
            exec.shutdownNow();
        }
    }
    
    public void execute(Runnable command) {
        execute(command, true);
    }

    public Future<?> execute(Runnable command, boolean isBackGrand) {
        try {
            if (isBackGrand) {
                exec.execute(command);
                return null;
            }
            else {
                return exec.submit(command);
            }
        }
        catch (Exception e) {
            LOGGER.error("async execute error", e);
        }
        return null;
    }

    /**
     * 异步执行，同时异步返回执行的结果，超时时抛出异常
     * @param callable Callable 线程
     * @param timeout 获得异步结果的超时时间，单位：秒
     * @throws TimeoutException 
     * @throws ExecutionException 
     * @throws InterruptedException 
     * @see
     */
    public Object futureTask(Callable<Object> callable, long timeout)
        throws InterruptedException, ExecutionException, TimeoutException {
        // 使用futureTask执行callable
        FutureTask<Object> futureTask = new FutureTask<>(callable);
        // 执行
        exec.execute(futureTask);
        // 等待返回结果，并返回到外层调用方法，设置超时时间
        return futureTask.get(timeout, TimeUnit.SECONDS);
    }

    @SuppressWarnings("unchecked")
    public <T> List<T> executeTask(TaskAction<T>... tasks) {
        final CountDownLatch latch = new CountDownLatch(tasks.length);

        List<Future<T>> futures = new ArrayList<Future<T>>();
        List<T> resultList = new ArrayList<>();

        for (final TaskAction<T> runnable : tasks) {
            Future<T> future = exec.submit(new Callable<T>() {
                @Override
                public T call()
                    throws Exception {
                    try {
                        return runnable.doInAction();
                    }
                    finally {
                        latch.countDown();
                    }
                }
            });
            futures.add(future);
        }

        try {
            latch.await();
        }
        catch (InterruptedException e) {
            LOGGER.error("Executing Task is interrupt.", e);
        }

        for (Future<T> future : futures) {
            try {
                T result = future.get();
                if (result != null) {
                    resultList.add(result);
                }
            }
            catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        return resultList;
    }

}
